package com.javaspeak.concurrency.pattern.actor;

import java.lang.reflect.Proxy;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import com.javaspeak.concurrency.pattern.application.UtilityActor;

/**
 * @author John Dickerson
 */
public class ActorRegistryImpl implements ActorRegistry {

    public static ActorRegistry instance;

    static {
        instance = new ActorRegistryImpl();
    }

    private Logger logger = Logger.getLogger( ActorRegistryImpl.class );

    private ExecutorService executorService = Executors.newCachedThreadPool();

    private ConcurrentHashMap<Class<? extends Object>, ActorReferences>
        actorReferencesByActorClassMap =
            new ConcurrentHashMap<Class<? extends Object>, ActorReferences>();


    @SuppressWarnings("rawtypes")
    private Class[] getInterfaces( Class<? extends Actor> actorClass ){

        Class[] directInterfaces = actorClass.getInterfaces();

        Class[] interfaces = new Class[ directInterfaces.length + 1 ];

        for ( int i=0; i<directInterfaces.length; i++ ){

            interfaces[ i ] = directInterfaces[ i ];
        }

        interfaces[ directInterfaces.length ] = Actor.class;

        return interfaces;
    }


    /**
     * Private constructor
     */
    private ActorRegistryImpl(){

    }


    public static ActorRegistry getInstance(){

        return instance;
    }


    /* (non-Javadoc)
     * @see com.javaspeak.actor.ActorRegistry#registerActor(
     *     com.javaspeak.actor.Actor)
     */
    public void registerActor( Actor actor ) {

        BlockingQueue<MethodRequest> methodRequestBlockingQueue =
            new LinkedBlockingQueue<MethodRequest>();

        Class<? extends Actor> actorClass = actor.getClass();

        logger.debug( "actorClass = " + actorClass );

        @SuppressWarnings("unchecked")
        Class<? extends Actor>[] actorInterfaces =
                getInterfaces( actor.getClass() );

        Actor actorProxy =
            ( Actor )Proxy.newProxyInstance(
                UtilityActor.class.getClassLoader(),
                    actorInterfaces, new ActorInvocationHandler(
                            methodRequestBlockingQueue, actor ) );

        for ( int i=0; i<8; i++ ){

            ProcessMethodsRunnable processMethodsRunnable =
                new ProcessMethodsRunnable( methodRequestBlockingQueue );

            executorService.submit( processMethodsRunnable );
        }

        ActorReferences actorReferences =
            new ActorReferencesImpl( actor, actorProxy );

        actorReferencesByActorClassMap.put(
                actorClass.getInterfaces()[ 0 ], actorReferences );
    }


    /* (non-Javadoc)
     * @see com.javaspeak.actor.ActorRegistry#getActor(java.lang.Class)
     */
    public Actor getActor( Class<? extends Object> actorClass ) {

        ActorReferences actorReferences =
            actorReferencesByActorClassMap.get( actorClass );

        Actor proxiedActor = actorReferences.getProxiedActor();

        return proxiedActor;
    }



    /* (non-Javadoc)
     * @see com.javaspeak.actor.ActorRegistry#shutDown()
     */
    public void shutDown() {

        executorService.shutdownNow();
    }
}
